package org.example


import org.apache.spark.sql.SparkSession

import java.util.regex.Pattern

object data1_moives {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder
      .master("local[*]")
      .appName("spark")
      .getOrCreate()
    val sc = spark.sparkContext
//    电影数据分析
    val filePath="src/main/resources/"
    val moviesRDD=sc.textFile(filePath+"movies.dat")
    val occupationsRDD=sc.textFile(filePath+"occupations.dat")
    val ratingsRDD=sc.textFile(filePath+"ratings.dat")
    val usersRDD=sc.textFile(filePath+"users.dat")
    println("电影总数："+moviesRDD.count()+"\n"+"用户总数"+usersRDD.count())
    println("评分总数:"+ratingsRDD.count()+"\n"+"职业数"+occupationsRDD.count())
//    分析一：统计某用户评分的电影总数以及对应的电影详情（id,电影名，类型）
//   val num= ratingsRDD.map(_.split("::"))
//      .map(user => (user(1),user(0)))//转换成key value 电影id 用户id
    //    用户id为3088的评分电影总数
//    .filter(_._2.equals("3088"))
//       println("用户id为3088的评分电影总数："+num.count())
//    所评分电影对应的详情(电影名，电影类型)
//    val moviesInfo=moviesRDD.map(_.split("::"))
//      .map(movie =>(movie(0),(movie(1),movie(2))))
//    val res = num.join(moviesInfo) //(MovieId,(UserId,(Title,Genres))
//      .map(item =>(item._1,item._2._1,item._2._2._1,item._2._2._2))
//    println("电影详情是：")
//    res.take(3).foreach(println)
//
//   分析二：找出评分最高的前10部电影名，以及观影人数最多的前10部电影
    // "ratings.dat"：UserID::MovieID::Rating::Timestamp
    // [UserID,MovieID,Rating,Timestamp]
//     val rating = ratingsRDD.map(x => x.split("::")).map {  x => {    (x(0), x(1), x(2))
////       (UserID,MovieID,Rating)
//      //
//             }
//             }.cache()
//    println("平均得分最高的前 10 名的电影名称简单版")
//     //  ( MovieId,( Rating, 1) )  (1200,(4.0,1))
//      rating.map(x => (x._2, (x._3.toDouble, 1)))
//    //  ( MovieId,( 总分, 总次数) )
//      .reduceByKey((x, y) => {    (x._1 + y._1, x._2 + y._2)   })
//    // (  平均分, MovieId )
//     .map(x => (x._2._1 / x._2._2, x._1))
//     .sortByKey(false)
//     .take(5)
//      .foreach(println)
//    println("按平均分取前 10 部电影输出详情:(平均分,(movieId,Title,Genres,总分,总次 数))")
//     // MovieID::Title::Genres
//     val moviesInfo = moviesRDD.map(x => x.split("::"))
//       .map(x => {    (x(0), (x(1), x(2)))   })
//
//    val ratingsInfo = rating.map(x => (x._2, (x._3.toDouble, 1)))
//    //  ( MovieId,( Rating, 1) )  (1252,(4.0,1))
//      .reduceByKey((x, y) => {    (x._1 + y._1, x._2 + y._2)   }) //  ( MovieId,( 总分, 总次数) )
//     .map(x => (x._1, (x._2._1 / x._2._2, x._2._1, x._2._2))) //   ( MovieId, (平均分, 总分,总次数) )
//    moviesInfo.join(ratingsInfo)
//      .map(info => {
//         (info._2._2._1, (info._1, info._2._1._1, info._2._1._2, info._2._2._2, info._2._2._3))
//       // (平均分,(movieId,Title,Genres,总分,总次数))
//        }).sortByKey(false)
//        .take(5)
//        .foreach(println)
//      println("观影人数最多的前 10 部电影")
//       // (UserID,MovieID,Rating)
//       val watchViewsInfo = rating.map(x => {
//      (x._2, 1)
//
//    }).reduceByKey((x, y) => x + y) //  ( MovieId,总次数 )
//       .map(x => (x._2, x._1))
//         .sortByKey(false)
//         .take(5)
//      // 5 名
//       watchViewsInfo.foreach(println(_))
//      println("===================>")
//    rating.map(x => (x._2, 1)) //  ( MovieId, 1)
//       .reduceByKey((x, y) => {    (x + y)   })
//      //  ( MovieId,总次数 )
//        .map(x => (x._2, x._1)) //   (  总次数, MovieId )
//        .sortByKey(false)
//      .take(10)   .foreach(println) //  286-> 999
//    println("详情的输出(  观影人数，电影编号)")
//     // 输出 ( 总次数, (MovieID, title,Genres,总分,平均分  ) ) //  ratingsInfo   ( MovieId, (平均分, 总分,总次数) ) //  ( MovieID,  (  (Title,Genres),(平均分, 总分,总次数)  ) ) // ratingsInfo ( MovieId, (平均分, 总分,总次数) ) // moviesInfo  MovieID::Title::Genres
//       moviesInfo.join(ratingsInfo).map(x => {
//         (x._2._2._3, (x._2._1._1, x._2._1._2, x._2._2._1, x._2._2._1))
//       }).sortByKey(false)   .take(10)   .foreach(println)
//    分析三：统计不同电影类型数量
      val moviesStyle = moviesRDD.map(_.split("::"))
        .map(tp => (tp(0),tp(2))) //只留电影的编号和类型
        .flatMapValues(sp => sp.split("\\|"))  //切分电影类型
        .map(p =>(p._2,1))//转换成键值对的形式
        .reduceByKey((x,y)=>x+y)
  //    打印喜剧片（comedy）的数据
         .filter(_._1.equals("Comedy"))
    moviesStyle.foreach(println)
    //    moviesStyle.take(5).foreach(println)
//    分析四：每个年份电影拍摄的总数
//    正则表达式
//       val pattern = Pattern.compile("(.*)(\\(\\d{4}\\))")
//       val movieYear=moviesRDD.map(_.split("::"))
//         .map(x =>(x(1),1))
//         .map(x=>{
//           var name = ""
//           var year = ""
//           val matcher =pattern.matcher(x._1)
//           if(matcher.find()){
//             name=matcher.group(1)
//             year=matcher.group(2)
//             year = year.substring(1,year.length-1)
//           }
//           if(year == ""){
//             (-1,1)//年份为空给个标识码-1
//           }else{}
//           (year.toInt,1)//不为空写成key value
//         })
//         .reduceByKey((x,y)=>x+y)
////        打印喜剧片的电影数量和2000年度生产的电影数量
//          .filter(_._1==2000).map(_._2)
//    movieYear.foreach(println)
//    需求四
    // "ratings.dat"：UserID::MovieID::Rating::Timestamp
    // [UserID,MovieID,Rating,Timestamp]
     val rating = ratingsRDD.map(x => x.split("::")).map {   x => {    // (UserID,MovieID,Rating)
        (x(0), (x(1), x(2)))    }   }.cache()
     // "users.dat"：UserID::Gender::Age::OccupationID::Zip-code
     val usersInfo = usersRDD.map(x => x.split("::"))   // (UserID,Gender)
     .map(x => (x(0), x(1)))    .cache()
    println("男性总共有：", usersInfo.filter(x => x._2.equals("M")).count())
    println("女性总共有：", usersInfo.filter(x => x._2.equals("F")).count())
     // (UserID,((MovieID,Rating),Gender))
     val data = rating.join(usersInfo)    .map(x => {    // (UserID,MovieID,Rating,Gender)
          (x._1, x._2._1._1, x._2._1._2, x._2._2)    })
    val femaleRatings = data.filter(x => x._4.equals("F")).cache()
    val maleRatings = data.filter(x => x._4.equals("M")).cache()
    println("女性评分总分最高的 10 部电影, 格式：总评分,电影 ID")
     // (UserID,MovieID,Rating,Gender)
      femaleRatings
     // (MovieID,Rating)
      .map(x => (x._2, x._3.toDouble))
      // (MovieID,总评分)
       .reduceByKey((x, y) => x + y)
    .map(x => (x._2, x._1))
    .sortByKey(false)
    .take(10)
    .foreach(println(_))
    println("男性评分总分最高的 10 部电影, 格式：总评分,电影 ID")
     // (UserID,MovieID,Rating,Gender)
     maleRatings
    // (MovieID,Rating)
     .map(x => (x._2, x._3.toDouble))
    // (MovieID,总评分)
     .reduceByKey((x, y) => x + y)    .map(x => (x._2, x._1))    .sortByKey(false)    .take(10)    .foreach(println(_))
    // 1. 分析男性用户最喜欢看的前 10 部电影
    // 1. 女性用户最喜欢看的前 10 部电影  输出格式:  (movieId, 电影名,总分，打分次数，平均 分)
    // 1. 分析最受不同年龄段人员欢迎的 前 10 部电影
    // 附加显示电影的详细信息
    // 累计总分后计算平均分，按平均分排序，显示 top 10,显示电影名  并且显示电影的详细信息
    // (UserID,MovieID,Rating,Gender)
     val femaleRatingsDetail = femaleRatings.map(x => {   // (movieId,(rating,1))
        (x._2, (x._3.toDouble, 1))   })
    // (movieId,(总评分,总评次数))
     .reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2))
    // (movieId,(总评分,总次数,平均分))
     .map(x => {     (x._1, (x._2._1, x._2._2, x._2._1 / x._2._2))    })
    // (UserID,MovieID,Rating,Gender)
     val maleRatingsDetail = maleRatings.map(x => {   // (movieId,(rating,1))
        (x._2, (x._3.toDouble, 1))   })
    // (movieId,(总评分,总评次数))
     .reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2))
    // (movieId,(总评分,总次数,平均分))
     .map(x => {     (x._1, (x._2._1, x._2._2, x._2._1 / x._2._2))    })
     // MovieID::Title::Genres
     val movieInfo = moviesRDD.map(x => x.split("::"))   // (MovieID,(Title,Genres))
      .map(x => (x(0), (x(1), x(2))))
    println("============================")

     // (MovieID,((总评分,总次数,平均分),(Title,Genres))  femaleRatingsDetail.join(movieInfo)    .map(x => {    // (平均分,(MovieID,总评分,总次数,Title,Genres))     (x._2._1._3, (x._1, x._2._1._1, x._2._1._2, x._2._2._1, x._2._2._2))    })    .sortByKey(false)    .take(10)    .foreach(println)
    println("男性评分平均分最高的 10 部电影的详情")
     // ( MovieID,((总评分,总次数,平均分),(Title,Genres))  maleRatingsDetail.join(movieInfo)    .map(x => {     (x._2._1._3, (x._1, x._2._1._1, x._2._1._2, x._2._2._1, x._2._2._2))    }) //( 平均分,(MovieID, 总评分,总次数, Title,Genres ) )    .sortByKey(false)    .take(10)    .foreach(println)
//需求五


    sc.stop()
  }
  }
